package analysis_uv;

import beans.PageViewCount;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.runtime.operators.util.BloomFilter;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashSet;

/**
 * @author zkq
 * @date 2022/10/4 20:52
 */
public class UniqueVisitor {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<UserBehavior> inputStream = env
                .readTextFile("F:\\javasecode220620\\UserBehaviorAnalysis\\NetworkFlowAnalysis\\src\\main\\resources\\UserBehavior.csv")
                .map(data -> {
                    String[] splits = data.split(",");
                    return new UserBehavior(Long.parseLong(splits[0]), Long.parseLong(splits[1]),Integer.parseInt(splits[2]),
                            splits[3], Long.parseLong(splits[4]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                            @Override
                            public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                return element.getTimestamp() * 1000;
                            }
                        })
                );
        //开窗统计uv 1小时 利用set去重 用windowall去实现 有更好方案 此方案的缺点为 windowall并行度只能为1
        //且使用了全窗口函数 等到数据到齐了才会处理 可以使用增量+全量方案
        //且消耗内存 因为把userId全部存在set里了 userId足够多的就麻烦了
        SingleOutputStreamOperator<PageViewCount> result = inputStream
                .filter(data -> "pv".equals(data.getBehavior()))
                .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
                .process(new ProcessAllWindowFunction<UserBehavior, PageViewCount, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<UserBehavior> elements, Collector<PageViewCount> out) throws Exception {
                        //存userId
                        HashSet<Long> users = new HashSet<>();
                        for (UserBehavior element : elements) {
                            users.add(element.getUserId());
                        }
                        out.collect(new PageViewCount("uv", context.window().getEnd(), (long) users.size()));
                    }
                });
        result.print();
        env.execute("uv job");
    }
}
